package mr

import (
	"fmt"
	"hash/fnv"
	"io/ioutil"
	"log"
	"net/rpc"
	"os"
	"sort"
	"strings"
)

type ByKey []KeyValue

// for sorting by key.
func (a ByKey) Len() int           { return len(a) }
func (a ByKey) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
func (a ByKey) Less(i, j int) bool { return a[i].Key < a[j].Key }

//
// Map functions return a slice of KeyValue.
//
type KeyValue struct {
	Key   string
	Value string
}

//
// use ihash(key) % NReduce to choose the reduce
// task number for each KeyValue emitted by Map.
//
func ihash(key string) int {
	h := fnv.New32a()
	h.Write([]byte(key))

	return int(h.Sum32() & 0x7fffffff)
}

func shuffle(kv []KeyValue, R int) [][]KeyValue {
	kvs := make([][]KeyValue, R)
	for _, k := range kv {
		kvs[ihash(k.Key)%R] = append(kvs[ihash(k.Key)%R], k)
	}
	return kvs
}

func workerMapTask(reply *GetReply, mapf func(string, string) []KeyValue, r int) {
	filename := reply.Task.Loc
	if filename == "" {
		return
	}
	file, err := os.Open(filename)
	if err != nil {

		log.Printf("cannot open %v", filename)
	}
	content, err := ioutil.ReadAll(file)
	if err != nil {
		log.Printf("cannot read %v", filename)
	}
	file.Close()
	kva := mapf(filename, string(content))

	// shuffle
	intermediate := shuffle(kva, r)
	// sort
	// write
	for i, kv := range intermediate {

		oname := fmt.Sprintf("mr-%d-%d", reply.Task.ID, i)
		onametmp := oname + "tmp"
		interfile, _ := os.Create(onametmp)
		for _, v := range kv {
			fmt.Fprintf(interfile, "%v %v\n", v.Key, v.Value)
		}
		interfile.Close()
		os.Rename(onametmp, oname)
	}

}
func workerReduceTask(reply *GetReply, reducef func(string, []string) string) {
	files := reply.Task.Loc
	if files == "" {
		return
	}
	intermediate := []KeyValue{}
	for _, filename := range strings.Split(files, ";") {
		if filename != "" {
			file, err := os.Open(filename)
			if err != nil {
				log.Println(err)
				log.Printf("cannot open %v", filename)
			}
			content, _ := ioutil.ReadAll(file)
			for _, v := range strings.Split(string(content), "\n") {
				ss := strings.Split(v, " ")
				if len(ss) == 2 {
					intermediate = append(intermediate, KeyValue{
						Key:   ss[0],
						Value: ss[1],
					})
				}
			}

		}
	}
	sort.Sort(ByKey(intermediate))
	i := 0

	oname := fmt.Sprintf("mr-out-%d", reply.Task.ID)
	onametmp := oname + "tmp"
	ofile, _ := os.Create(onametmp)
	for i < len(intermediate) {
		j := i + 1
		for j < len(intermediate) && intermediate[j].Key == intermediate[i].Key {
			j++
		}
		values := []string{}
		for k := i; k < j; k++ {
			values = append(values, intermediate[k].Value)
		}
		output := reducef(intermediate[i].Key, values)

		// this is the correct format for each line of Reduce output.
		fmt.Fprintf(ofile, "%v %v\n", intermediate[i].Key, output)

		i = j
	}
	ofile.Close()

	os.Rename(onametmp, oname)
}

//
// main/mrworker.go calls this function.
//
func Worker(mapf func(string, string) []KeyValue,
	reducef func(string, []string) string) {

	// Your worker implementation here.
	for {
		worker(mapf, reducef)
	}
}

func worker(mapf func(string, string) []KeyValue,
	reducef func(string, []string) string) {

	reply := &GetReply{}
	callGetIdleTask(reply)
	switch {
	case reply.Task.Style == Map:
		workerMapTask(reply, mapf, reply.NReduce)
	case reply.Task.Style == Reduce:
		workerReduceTask(reply, reducef)
	}
	args := &PutArgs{
		Task: reply.Task,
	}
	callPutTask(args)

}

func callGetIdleTask(reply *GetReply) {

	if !call("Coordinator.GetIdleTask", &GetArgs{}, reply) {
		os.Exit(0)
	}

	if reply.Task.Loc != "" {
		// log.Printf("[callGetIdleTask] id: %d style: %d loc: %s\n", reply.Task.ID, reply.Task.Style, reply.Task.Loc)
	}

}

func callPutTask(args *PutArgs) {
	if args.Task.Loc != "" {
		if !call("Coordinator.PutTask", args, &PutReply{}) {
			os.Exit(1)
		}
		if args.Task.Loc != "" {
			// log.Printf("[callPutTask] id: %d, style: %d, loc: %s\n", args.Task.ID, args.Task.Style, args.Task.Loc)
		}
	}

}

//
// example function to show how to make an RPC call to the coordinator.
//
// the RPC argument and reply types are defined in rpc.go.
//
func CallExample() {

	// declare an argument structure.
	args := ExampleArgs{}

	// fill in the argument(s).
	args.X = 99

	// declare a reply structure.
	reply := ExampleReply{}

	// send the RPC request, wait for the reply.
	call("Coordinator.Example", &args, &reply)

	// reply.Y should be 100.
	fmt.Printf("reply.Y %v\n", reply.Y)
}

//
// send an RPC request to the coordinator, wait for the response.
// usually returns true.
// returns false if something goes wrong.
//
func call(rpcname string, args interface{}, reply interface{}) bool {
	// c, err := rpc.DialHTTP("tcp", "127.0.0.1"+":1234")
	sockname := coordinatorSock()
	c, err := rpc.DialHTTP("unix", sockname)
	if err != nil {
		log.Fatal("dialing:", err)
	}
	defer c.Close()

	err = c.Call(rpcname, args, reply)
	if err == nil {
		return true
	}

	fmt.Println(err)
	return false
}
